package org.seed.quantum.message.rpc;

import org.seed.quantum.message.plugins.Plugin;
import org.seed.quantum.socket.AbstractSocketSession;
import org.seed.quantum.socket.MessageProcessor;
import org.seed.quantum.socket.NetMonitor;
import org.seed.quantum.socket.SocketStateEnum;
import java.io.*;
import java.lang.reflect.Proxy;
import java.net.SocketTimeoutException;
import java.nio.channels.AsynchronousSocketChannel;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;

public class RpcConsumerProcessor implements MessageProcessor<byte[]>, NetMonitor {


    private Map<String, CompletableFuture<RpcResponse>> syncRespMap = new ConcurrentHashMap<>();
    private Map<Class, Object> objectMap = new ConcurrentHashMap<>();
    private AbstractSocketSession aioSession;


    @Override
    public void process(AbstractSocketSession session, byte[] msg) {
        ObjectInput objectInput = null;
        try {
            objectInput = new ObjectInputStream(new ByteArrayInputStream(msg));
            RpcResponse resp = (RpcResponse) objectInput.readObject();
            syncRespMap.get(resp.getUuid()).complete(resp);
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            if (objectInput != null) {
                try {
                    objectInput.close();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        }

    }

    public <T> T getObject(final Class<T> remoteInterface) {
        Object obj = objectMap.get(remoteInterface);
        if (obj != null) {
            return (T) obj;
        }
        obj = (T) Proxy.newProxyInstance(getClass().getClassLoader(), new Class[]{remoteInterface},
                (proxy, method, args) -> {
                    RpcRequest req = new RpcRequest();
                    req.setInterfaceClass(remoteInterface.getName());
                    req.setMethod(method.getName());
                    Class<?>[] types = method.getParameterTypes();
                    if (types != null && types.length > 0) {
                        String[] paramClass = new String[types.length];
                        for (int i = 0; i < types.length; i++) {
                            paramClass[i] = types[i].getName();
                        }
                        req.setParamClassList(paramClass);
                    }
                    req.setParams(args);
                    RpcResponse rmiResp = sendRpcRequest(req);
                    if (rmiResp.getException() != null) {
                        throw new RuntimeException(rmiResp.getException());
                    }
                    return rmiResp.getReturnObject();
                });
        objectMap.put(remoteInterface, obj);
        return (T) obj;
    }

    private final RpcResponse sendRpcRequest(RpcRequest request) throws Exception {
        CompletableFuture<RpcResponse> rpcResponseCompletableFuture = new CompletableFuture<>();
        syncRespMap.put(request.getUuid(), rpcResponseCompletableFuture);

        //输出消息
        ByteArrayOutputStream byteArrayOutputStream = new ByteArrayOutputStream();
        ObjectOutput objectOutput = new ObjectOutputStream(byteArrayOutputStream);
        objectOutput.writeObject(request);
        byte[] data = byteArrayOutputStream.toByteArray();
        synchronized (aioSession) {
            aioSession.writeBuffer().writeInt(data.length + 4);
            aioSession.writeBuffer().write(data);
            aioSession.writeBuffer().flush();
        }

        try {
            RpcResponse resp = rpcResponseCompletableFuture.get(7, TimeUnit.SECONDS);
            return resp;
        } catch (Exception e) {
            throw new SocketTimeoutException("rpc is timeout!");
        }
    }

    @Override
    public void stateEvent(AbstractSocketSession session, SocketStateEnum socketStateEnum, Throwable throwable) {
        switch (socketStateEnum) {
            case NEW_SESSION:
                setAioSession(session);
                break;
        }
    }

    public void setAioSession(AbstractSocketSession aioSession) {
        this.aioSession = aioSession;
    }

    private final List<Plugin<byte[]>> plugins = new ArrayList<>();

    @Override
    public final void afterRead(AbstractSocketSession session, int readSize) {
        for (Plugin<byte[]> plugin : plugins) {
            plugin.afterRead(session, readSize);
        }
    }

    @Override
    public final void afterWrite(AbstractSocketSession session, int writeSize) {
        for (Plugin<byte[]> plugin : plugins) {
            plugin.afterWrite(session, writeSize);
        }
    }

    @Override
    public final void beforeRead(AbstractSocketSession session) {
        for (Plugin<byte[]> plugin : plugins) {
            plugin.beforeRead(session);
        }
    }

    @Override
    public final void beforeWrite(AbstractSocketSession session) {
        for (Plugin<byte[]> plugin : plugins) {
            plugin.beforeWrite(session);
        }
    }

    @Override
    public final AsynchronousSocketChannel shouldAccept(AsynchronousSocketChannel channel) {
        AsynchronousSocketChannel acceptChannel = channel;
        for (Plugin<byte[]> plugin : plugins) {
            acceptChannel = plugin.shouldAccept(acceptChannel);
            if (acceptChannel == null) {
                return null;
            }
        }
        return acceptChannel;
    }

    public final void addPlugin(Plugin<byte[]> plugin) {
        this.plugins.add(plugin);
    }

}
